package com.ruoyi.utils;

import com.ruoyi.common.utils.SseEmitterUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.*;
import java.nio.file.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 监控文件,并输出 append的内容
 */
@Slf4j
@Component
public class FileLogMonitor {

    public void cancel(String filePath) {
        WatchService watchService = monitorCache.get(filePath);
        if (null != watchService) {
            try {
                watchService.close();
            }catch (Exception e) {
                log.error("",e);
            }
        }
    }

    /**
     * 判断是否日志监控完,最多等待3s
     * @param filePath
     */
    public void finishLog(String filePath) {
        int num = 1;
        while (true) {
            WatchService watchService = monitorCache.get(filePath);
            if (null != watchService) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    //
                }
            }else {
                log.info("日志结束:",filePath);
                break;
            }
            if (num == 3) {
                log.info("日志超时退出结束:",filePath);
                break;
            }
            num++;
        }
    }

    private Map<String,WatchService> monitorCache = new ConcurrentHashMap();

    /**
     * 监听某个文件,并用sse输出至页面...页面需先订阅sse
     * @param filePath
     * @param sseId
     * @throws Exception
     */
    @Async("asyncExecutor")
    public void monitor(String filePath, String sseId)  {

        try {
            asyncMoitor(filePath,sseId);
        }catch (Exception e) {
            log.error("监听文件异常:{}",filePath,e);
        }
    }

    /**
     * 监听某个文件,并用sse输出至页面...一个页面建立多个SSE链接的情况 需要将消息包装一下
     * @param filePath
     * @param sseId
     * @param type 反演评估fy  建模评估 jm
     * @throws Exception
     */
    @Async("asyncExecutor")
    public void monitor(String filePath, String sseId,String type)  {

        try {
            asyncMoitorType(filePath,sseId,type);
        }catch (Exception e) {
            log.error("监听文件异常:{}",filePath,e);
        }
    }

    public void syncMonitor(String filePath, String sseId,String type) throws Exception {

        // 创建一个文件系统监听器
        WatchService watchService = FileSystems.getDefault().newWatchService();
        //放入缓存
        monitorCache.put(filePath,watchService);
        // 获取文件所在的目录路径
        Path dir = Paths.get(filePath).getParent();
        // 注册监听器来监听文件变化
        dir.register(watchService, StandardWatchEventKinds.ENTRY_MODIFY);
        log.info("Watching file for changes: {}",filePath);

        // 记录文件当前的长度
        long lastPosition = 0;
        log.debug("输出已有的");
        //先输出文件已有的信息
        try (RandomAccessFile file = new RandomAccessFile(filePath, "r")) {
            file.seek(lastPosition);

            try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file.getFD()), "UTF-8"))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    // 在这里可以对读取到的内容进行处理
                    if (StringUtils.hasLength(line)) {
                        SseEmitterUtils.sendMessage(sseId, type+":"+line);
                    }
                }

                // 更新记录的位置为文件的当前长度
                lastPosition = file.getFilePointer();
            }
        } catch (Exception ex) {
            log.error("",ex);
        }


        //开始监听文件修改的内容
        A: while  (true) {
            WatchKey key;
            try {
                //超过20分钟 自动停止
                key = watchService.poll(20, TimeUnit.MINUTES);
            } catch (InterruptedException e) {
                log.info("线程终止:{}",filePath);
                break A;
            }
            if (null == key) {
                log.info("monitor 超时 自动跳出: {}",filePath);
                break A;
            }

            for (WatchEvent<?> event : key.pollEvents()) {
                WatchEvent.Kind<?> kind = event.kind();

                // 确保是文件修改事件
                if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
                    Path modifiedFilePath = (Path) event.context();
                    if (modifiedFilePath.endsWith(Paths.get(filePath).getFileName())) {
                        log.debug("监听到修改");
                        // 打开文件并定位到上次记录的位置 暂停一会儿,让写操作写完
                        Thread.sleep(1000);
                        try (RandomAccessFile file = new RandomAccessFile(filePath, "r")) {
                            file.seek(lastPosition);

                            try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file.getFD()), "UTF-8"))) {
                                String line;
                                while ((line = reader.readLine()) != null) {
                                    // 在这里可以对读取到的内容进行处理
                                    if (StringUtils.hasLength(line)) {
                                        SseEmitterUtils.sendMessage(sseId, type+":"+line);
                                    }
                                }

                                // 更新记录的位置为文件的当前长度
                                lastPosition = file.getFilePointer();
                            }
                        } catch (IOException ex) {
                            log.error("",ex);
                        }
                    }
                }
            }

            // 重置监听键，准备下一次监听
            boolean valid = key.reset();
            if (!valid) {
                break;
            }
        }
        log.info("文件监听结束:{}",filePath);
        watchService.close();
        //放入缓存
        monitorCache.remove(filePath);
        SseEmitterUtils.sendMessage(sseId, type+":600");
        SseEmitterUtils.remove(sseId);
    }

    private void asyncMoitor(String filePath, String sseId) throws Exception {
        // 创建一个文件系统监听器
        WatchService watchService = FileSystems.getDefault().newWatchService();
        //放入缓存
        monitorCache.put(filePath,watchService);
        // 获取文件所在的目录路径
        Path dir = Paths.get(filePath).getParent();
        // 注册监听器来监听文件变化
        dir.register(watchService, StandardWatchEventKinds.ENTRY_MODIFY);
        log.info("Watching file for changes: {}",filePath);

        // 记录文件当前的长度
        long lastPosition = 0;
        //先输出文件已有的信息
        log.debug("输出已有的");
        try (RandomAccessFile file = new RandomAccessFile(filePath, "r")) {
            file.seek(lastPosition);

            try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file.getFD()), "UTF-8"))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    // 在这里可以对读取到的内容进行处理
                    if (StringUtils.hasLength(line)) {
                        SseEmitterUtils.sendMessage(sseId, line);
                        if ("==END==".equals(line.trim())) {
                            log.info("文件监听结束:{}",filePath);
                            SseEmitterUtils.sendMessage(sseId, "600");
                            SseEmitterUtils.remove(sseId);
                            monitorCache.remove(filePath);
                            return;
                        }
                    }
                }

                // 更新记录的位置为文件的当前长度
                lastPosition = file.getFilePointer();
            }
        } catch (Exception ex) {
            log.error("",ex);
        }

        //开始监听文件修改的内容
        A: while  (true) {
            WatchKey key;
            try {
                //超过2分钟 自动停止
                key = watchService.poll(2, TimeUnit.MINUTES);
            } catch (InterruptedException e) {
                log.info("线程终止:{}",filePath);
                break A;
            }
            if (null == key) {
                log.info("monitor 超时 自动跳出: {}",filePath);
                break A;
            }

            for (WatchEvent<?> event : key.pollEvents()) {
                WatchEvent.Kind<?> kind = event.kind();

                // 确保是文件修改事件
                if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
                    Path modifiedFilePath = (Path) event.context();
                    if (modifiedFilePath.endsWith(Paths.get(filePath).getFileName())) {
                        // 打开文件并定位到上次记录的位置 暂停一会儿,让写操作写完
                        log.debug("检测到修改");
                        Thread.sleep(1000);
                        try (RandomAccessFile file = new RandomAccessFile(filePath, "r")) {
                            file.seek(lastPosition);

                            try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file.getFD()), "UTF-8"))) {
                                String line;
                                while ((line = reader.readLine()) != null) {
                                    // 在这里可以对读取到的内容进行处理
                                    if (StringUtils.hasLength(line)) {
                                        SseEmitterUtils.sendMessage(sseId, line);
                                        if ("==END==".equals(line.trim())) {
                                            //监测到终止
                                            log.info("monitor 监测到终止: {}",filePath);
                                            //600 终止标志
                                            SseEmitterUtils.sendMessage(sseId, "600");
                                            SseEmitterUtils.remove(sseId);
                                            watchService.close();
                                            break A;
                                        }
                                    }
                                }

                                // 更新记录的位置为文件的当前长度
                                lastPosition = file.getFilePointer();
                            }
                        } catch (IOException ex) {
                            log.error("",ex);
                        }
                    }
                }
            }

            // 重置监听键，准备下一次监听
            boolean valid = key.reset();
            if (!valid) {
                break;
            }
        }
        watchService.close();
        //放入缓存
        monitorCache.remove(filePath);
        log.info("文件监听结束:{}",filePath);
    }

    private void asyncMoitorType(String filePath, String sseId,String type) throws Exception {
        // 创建一个文件系统监听器
        WatchService watchService = FileSystems.getDefault().newWatchService();
        //放入缓存
        monitorCache.put(filePath,watchService);
        // 获取文件所在的目录路径
        Path dir = Paths.get(filePath).getParent();
        // 注册监听器来监听文件变化
        dir.register(watchService, StandardWatchEventKinds.ENTRY_MODIFY);
        log.info("Watching file for changes: {}",filePath);

        // 记录文件当前的长度
        long lastPosition = 0;
        log.debug("输出已有的");
        //先输出文件已有的信息
        try (RandomAccessFile file = new RandomAccessFile(filePath, "r")) {
            file.seek(lastPosition);

            try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file.getFD()), "UTF-8"))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    // 在这里可以对读取到的内容进行处理
                    if (StringUtils.hasLength(line)) {
                        SseEmitterUtils.sendMessage(sseId, type+":"+line);
                        if ("==END==".equals(line.trim())) {
                            log.info("文件监听结束:{}",filePath);
                            SseEmitterUtils.sendMessage(sseId, type+":600");
                            SseEmitterUtils.remove(sseId);
                            watchService.close();
                            monitorCache.remove(filePath);
                            return;
                        }
                    }
                }

                // 更新记录的位置为文件的当前长度
                lastPosition = file.getFilePointer();
            }
        } catch (Exception ex) {
            log.error("",ex);
        }


        //开始监听文件修改的内容
        A: while  (true) {
            WatchKey key;
            try {
                //超过60s 自动停止
                key = watchService.poll(120, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                // 可以记录错误信息或者直接终止程序
                log.info("线程终止:{}",filePath);
                break A;
            }
            if (null == key) {
                log.info("monitor 超时 自动跳出: {}",filePath);
                break A;
            }

            for (WatchEvent<?> event : key.pollEvents()) {
                WatchEvent.Kind<?> kind = event.kind();

                // 确保是文件修改事件
                if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
                    Path modifiedFilePath = (Path) event.context();
                    if (modifiedFilePath.endsWith(Paths.get(filePath).getFileName())) {
                        log.debug("监听到修改");
                        // 打开文件并定位到上次记录的位置 暂停一会儿,让写操作写完
                        Thread.sleep(1000);
                        try (RandomAccessFile file = new RandomAccessFile(filePath, "r")) {
                            file.seek(lastPosition);

                            try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file.getFD()), "UTF-8"))) {
                                String line;
                                while ((line = reader.readLine()) != null) {
                                    // 在这里可以对读取到的内容进行处理
                                    if (StringUtils.hasLength(line)) {
                                        SseEmitterUtils.sendMessage(sseId, type+":"+line);
                                        if ("==END==".equals(line.trim())) {
                                            log.info("文件监听结束:{}",filePath);
                                            SseEmitterUtils.sendMessage(sseId, type+":600");
                                            SseEmitterUtils.remove(sseId);
                                            break A;
                                        }
                                    }
                                }

                                // 更新记录的位置为文件的当前长度
                                lastPosition = file.getFilePointer();
                            }
                        } catch (IOException ex) {
                            log.error("",ex);
                        }
                    }
                }
            }

            // 重置监听键，准备下一次监听
            boolean valid = key.reset();
            if (!valid) {
                break;
            }
        }
        watchService.close();
        //放入缓存
        monitorCache.remove(filePath);
        log.info("文件监听结束:{}",filePath);
    }
}
